package com.example.springboottest.example.zookeeper.listener;

import com.example.springboottest.example.zookeeper.ZKclient;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

/**
 * 实现对 ZooKeeper 服务器端的事件监听，是客户端操作服务器的一项重点工作。在 Curator 的
 * API 中，事件监听有两种模式：
 * （1）一种是标准的观察者模式；
 * （2）另一种是缓存监听模式。
 * 第一种标准的观察者模式是通过 Watcher 监听器实现的；第二种缓存监听模式是通过引入了一
 * 种本地缓存视图 Cache 机制去实现的。
 * 第二种 Cache 事件监听机制，可以理解为一个本地缓存视图与远程 ZooKeeper 视图的对比过
 * 程，简单来说，Cache 在客户端缓存了 ZNode 的各种状态，当感知到 Zk 集群的 ZNode 状态变化时，
 * 会触发事件，注册的监听器会处理这些事件。
 * <p>
 * 需要引入
 * implementation 'org.apache.zookeeper:zookeeper:3.4.8'
 * <p>
 * Watcher监听器的注册是一次性的，当第二次改变节点内容时，注册已经失效，无法再次捕获节点的变动事件
 * 需要反复地通过构造者的 usingWatcher 方法去提前进行注册。
 */
public class ZkWatcherDemo {
    private String workerPath = "/test/listener/remoteNode";
    private String subWorkerPath = "/test/listener/remoteNode/id-";

    public static void main(String[] args) {
        new ZkWatcherDemo().testWatcher();
    }

    //利用 Watcher 来对节点进行监听操作
    public void testWatcher() {
        CuratorFramework client = ZKclient.instance.getClient();
        //检查节点是否存在，没有则创建
        boolean isExist = ZKclient.instance.isNodeExist(workerPath);
        if (!isExist) {
            ZKclient.instance.createNode(workerPath, null);
        }
        try {
            Watcher w = new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("监听到的变化 watchedEvent = " +
                            watchedEvent);

                }
            };
            while (true) {
                byte[] content = client.getData()
                        .usingWatcher(w).forPath(workerPath);
                System.out.println("监听节点内容：" + new String(content));
                // 第一次变更节点数据
//            client.setData().forPath(workerPath, "第 1 次更改内容".getBytes());
                // 第二次变更节点数据
//            client.setData().forPath(workerPath, "第 2 次更改内容".getBytes());
                Thread.sleep(5 * 1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
